/*
 * server.c
 *
 *  Created on: May 3, 2011
 *      Author: tvale, dmourao
 */
#include "common.h"
#include "server.h"
#include "queue.h"
#include <mpi.h>
#include <stdbool.h>
#include <stdlib.h>
#include <stdio.h>
#include <sys/time.h>
#include <string.h>

/*
 * Server state.
 *      queues: array of queues indexed by server;
 *      n_queues: # of queues;
 *      clock: local clock;
 *      var_local_copy: local copy of distributed variable;
 *      keepalive: self explainable;
 *      my_rank: self explainable;
 */
static queue_ptr*     queues;
static int            n_queues;
static clock_t        clock;
static int            var_local_copy;
static int            my_rank;
static FILE*          log_file_ptr;
static int            queues_finished;

/*
 * Operations.
 */
static void
log_open()
{
  char filename[100];

  sprintf(filename, "%s%i", "server", my_rank);

  log_file_ptr = fopen(filename, "w");
  if (log_file_ptr == NULL) {
    printf("Error opening %s for writing. Program terminated.", filename);
    abort();
  }
}

static void
log_close()
{
  fclose(log_file_ptr);
}

static void
log_write()
{
//  fprintf(log_file_ptr, "ts: %d, val: %d\n", clock.timestamp, var_local_copy);
  fprintf(log_file_ptr, "%d\n", var_local_copy);
  fflush(log_file_ptr);
}

static int
convert2index(int rank)
{
  return rank / 2;
}

static int
convert2rank(int index)
{
  return index * 2;
}

static bool
clock_newer(clock_t* c1, clock_t* c2)
{
  if(c1->timestamp > c2->timestamp ||
      (c1->timestamp == c2->timestamp && c1->rank > c2->rank))
    return true;

  return false;
}

static void
clock_update(clock_t* c)
{
  if (clock_newer(c, &clock))
    clock.timestamp = c->timestamp;
  clock.timestamp++;
}

static void
msg_send(int dst, msg_t* msg)
{
  MPI_Send(msg, 1, MPI_MESSAGE, dst, msg->type, MPI_COMM_WORLD);
  clock.timestamp++;
}

static void
msg_recv(msg_t* msg, MPI_Status* msg_status)
{
  MPI_Recv(msg, 1, MPI_MESSAGE, MPI_ANY_SOURCE,
      MPI_ANY_TAG, MPI_COMM_WORLD, msg_status);

  switch (msg_status->MPI_TAG) {
    case MSG_SET:
      clock.timestamp++;
      break;

    case MSG_ACK:
    case MSG_INQUIRE:
    case MSG_STOP_SRV:
      clock_update(&msg->payload.clock);
      break;

    case MSG_WRITE:
      clock_update(&msg->payload.write.clock);
      break;

    default:
      break;
  }
}

static void
msg_send_to_all(msg_t* msg)
{
  queue_add(queues[convert2index(my_rank)], *msg);

  for (int i = 0; i < n_queues; i++) {
    int dst = convert2rank(i);

    if (dst != my_rank)
      MPI_Send(msg, 1, MPI_MESSAGE, dst, msg->type, MPI_COMM_WORLD);
  }
}

static void
msg_process(msg_t* msg, MPI_Status* msg_status)
{
  msg_t   my_msg;
  write_t write;

  // chamar o msg_process_* correcto.
  switch(msg_status->MPI_TAG) {
    case MSG_GET_CLT:
      MPI_Send(&var_local_copy, 1, MPI_INT, msg_status->MPI_SOURCE,
               MSG_GET_SRV, MPI_COMM_WORLD);
      break;

    case MSG_SET:
      //clock.timestamp++;
      write.value          = msg->payload.value;
      write.clock          = clock;
      my_msg.type          = MSG_WRITE;
      my_msg.payload.write = write;

      msg_send_to_all(&my_msg);
      break;

    case MSG_STOP_CLT:
      //clock.timestamp++;
      my_msg.type          = MSG_STOP_SRV;
      my_msg.payload.clock = clock;

      msg_send_to_all(&my_msg);
      break;

    case MSG_INQUIRE:
      //clock.timestamp++;
      my_msg.type          = MSG_ACK;
      my_msg.payload.clock = clock;
      msg_send(msg_status->MPI_SOURCE, &my_msg);
      break;

    case MSG_WRITE:
    case MSG_ACK:
    case MSG_STOP_SRV:
      queue_add(queues[convert2index(msg_status->MPI_SOURCE)], *msg);
      break;

    default:
      break;
  }
}

static void
server_init(int rank, int n_servers)
{
  my_rank         = rank;
  n_queues        = n_servers;
  queues_finished = 0;

  // criar tantas filas quantos servidores
  queues = (queue_ptr*) calloc(n_queues, sizeof(queue_ptr));
  for (int i = 0; i < n_queues; i++)
    queues[i] = queue_create();

  // inicializar clock
  clock.rank      = my_rank;
  clock.timestamp = 0;

  log_open();
}

static void
queues_process()
{
  write_t*  oldest_write       = NULL;
  queue_ptr oldest_write_queue = NULL;

  for (int i = 0; i < n_queues; i++) {
    msg_t*    elem;
    queue_ptr queue     = queues[i];
    write_t*  candidate = NULL;

    if (queue_get_state(queue) != QUEUE_FINISHED) {
      elem = queue_peek(queue);

      switch (elem->type) {
        case MSG_STOP_SRV:
          queue_remove(queue);
          queue_set_state(queue, QUEUE_FINISHED);
          queues_finished++;
          break;

        case MSG_ACK:
          queue_remove(queue);
          break;

        case MSG_WRITE:
          candidate = &elem->payload.write;
          if (oldest_write == NULL ||
              clock_newer(&oldest_write->clock, &candidate->clock)) {
            oldest_write       = candidate;
            oldest_write_queue = queue;
          }
          break;

        default:
          break;
      }
    }
  }

  if (oldest_write != NULL) {
     var_local_copy = queue_remove(oldest_write_queue).payload.write.value;
     log_write();
   }
}


static void
print_queues()
{
  fprintf(log_file_ptr, "<queues>\n");
  for (int i = 0; i < n_queues; i++) {
    fprintf(log_file_ptr, "\t<server%i state=%i> -> ",
            convert2rank(i), queue_get_state(queues[i]));
    queue_print(queues[i], log_file_ptr);
    fprintf(log_file_ptr, "\n");
  }
  fprintf(log_file_ptr, "</queues>\n");
  fflush(log_file_ptr);
}


static void
queues_inquiry()
{
  for (int i = 0; i < n_queues; i++) {
    int dst = convert2rank(i);

    if (queue_get_state(queues[i]) == QUEUE_EMPTY && dst != my_rank) {
      clock.timestamp++;
      msg_t inq = {
          .type = MSG_INQUIRE,
          .payload = {
              .clock = clock
          }
      };
      msg_send(dst, &inq);
    }
  }
}

static void
msg_recv_all_needed()
{
  msg_t      msg;
  MPI_Status msg_status;
  bool       needs_msg = true;
  /*queue_ptr  my_queue = queues[convert2index(my_rank)];*/

  while (needs_msg) {
    needs_msg = false;
    /*fprintf(log_file_ptr, "before receive\n");
    print_queues();*/

    for (int i = 0; i < n_queues; i++) {
      queue_ptr queue = queues[i];
      if (queue_get_state(queue) == QUEUE_EMPTY) {
        needs_msg = true;
        break;
      }
    }

    if (needs_msg) {
      // receber uma mensagem.
      msg_recv(&msg, &msg_status);
      // tratar a mensagem.
      msg_process(&msg, &msg_status);
    }
  }
}

void
server_main(int rank, int n_servers)
{
  bool keepalive;

  server_init(rank, n_servers);

  keepalive = true;
  while (keepalive) {
    /*fprintf(log_file_ptr, "before queues_inquiry\n");
    print_queues();*/

    queues_inquiry();

    /*fprintf(log_file_ptr, "before msg_recv_all_needed\n");
    print_queues();*/

    msg_recv_all_needed();

    /*fprintf(log_file_ptr, "before queues_process\n");
    print_queues();*/

    queues_process();

    if (queues_finished == n_servers)
      keepalive = false;
  }

  fprintf(log_file_ptr, "terminated.");
  log_close();

  for (int i = 0; i < n_queues; i++)
    queue_destroy(queues[i]);
}

